%% RabbitMQ事件处理模块
%% 类似于Java中的一个事件处理器类，专门处理特定类型的事件。
-module(emqx_plugin_rabbitmq_evt).

%% 引入EMQX核心头文件
%% 类似于Java中的import语句，引入了EMQX框架的核心定义，比如 #message{} 记录的定义。
-include_lib("emqx/include/emqx.hrl").

%% 导出消息发布事件处理函数
%% 声明模块对外暴露的函数，类似于Java中public方法。
%% eventmsg_publish/1 表示函数名为eventmsg_publish，接受1个参数。
-export([
    eventmsg_publish/1
]).

%% 处理消息发布事件
%% 当EMQX接收到一条MQTT消息并准备发布时，会调用这个函数。
%% 参数Message: MQTT消息记录，这是一个Erlang的记录（record），类似于Java中的一个POJO类，包含了消息的各种属性。
%% #message{...} 是Erlang记录的模式匹配语法，用于解构参数，提取需要的字段值。
%% 返回: 包含事件相关信息的Map，Erlang的Map类似于Java中的HashMap。
eventmsg_publish(
    Message = #message{
        id = Id,                % 消息的唯一ID
        from = ClientId,        % 发送消息的客户端ID
        qos = QoS,              % 消息的服务质量等级 (0, 1, 2)
        flags = Flags,          % 消息的标志位 (e.g., retain flag)
        topic = Topic,          % 消息的主题
        payload = Payload,      % 消息的实际内容 (二进制数据)
        timestamp = Timestamp   % 消息在EMQX内部生成的时间戳
    }
) ->
    %% 调用 with_basic_columns 函数，进一步处理消息数据并添加一些基础信息。
    %% 'message.publish' 是一个原子（atom），类似于Java中的枚举或常量字符串，表示事件类型。
    with_basic_columns(
        'message.publish', % 事件名称，表明这是消息发布事件
        #{ % Erlang Map的语法，键值对结构
            id => emqx_guid:to_hexstr(Id), % 将消息ID转换为十六进制字符串
            clientid => ClientId, % 客户端ID
            username => emqx_message:get_header(username, Message, undefined), % 从消息头获取用户名，如果不存在则为undefined
            payload => Payload, % 消息体
            peerhost => ntoa(emqx_message:get_header(peerhost, Message, undefined)), % 获取客户端IP地址和端口，并转换为字符串
            topic => Topic, % 主题
            qos => QoS, % 服务质量等级
            flags => Flags, % 标志位
            pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})), % 处理MQTT 5.0的发布属性
            publish_received_at => Timestamp % 消息接收时间戳
        }
    ).

%% 添加基础列信息到事件数据中
%% 这个函数给处理后的消息数据Map添加一些通用的元数据。
%% 参数EventName: 事件名称 (原子类型)
%% 参数Columns: 原始列数据Map
%% when is_map(Columns) 是一个卫语句（guard），确保Columns参数确实是一个Map。
%% 返回: 添加了基础信息的新Map
with_basic_columns(EventName, Columns) when is_map(Columns) ->
    %% Columns#{ ... } 是更新Map的语法，将新的键值对添加到现有的Columns Map中。
    Columns#{ % 在原有Map基础上添加新的键值对
        event => EventName, % 事件名称
        timestamp => erlang:system_time(millisecond), % 当前系统时间戳（毫秒）
        node => node() % 当前Erlang节点的名称
    }.

%% 将IP地址转换为可读字符串
%% 这是一个辅助函数，用于将Erlang的IP地址表示形式转换成人类可读的字符串。
%% 参数可以是undefined、{IP,Port}元组或IP地址元组。
%% 返回binary类型的地址字符串，Erlang的binary类似于Java中的byte[]，但更常用于文本。
ntoa(undefined) -> undefined; % 如果输入是undefined，直接返回undefined
ntoa({IpAddr, Port}) -> % 如果输入是 {IP地址元组, 端口号}
    %% inet:ntoa/1 将IP地址元组转换为字符串
    %% integer_to_list/1 将整数端口号转换为字符串
    %% iolist_to_binary/1 将iolist（一个包含字符串、整数、二进制等的列表）高效地转换为二进制字符串
    iolist_to_binary([inet:ntoa(IpAddr), ":", integer_to_list(Port)]);
ntoa(IpAddr) -> % 如果输入仅是IP地址元组
    iolist_to_binary(inet:ntoa(IpAddr)).

%% 将消息头属性转换为可打印的Map格式
%% MQTT 5.0 引入了属性（Properties），这个函数用于将这些属性处理成更易读的Map格式。
%% 参数Headers: 消息头属性Map
%% 返回: 处理后的可打印Map
printable_maps(undefined) ->
    #{}; % 如果输入是undefined，返回一个空Map
printable_maps(Headers) ->
    %% maps:fold/3 类似于Java Stream API中的reduce操作，遍历Map的每个键值对并进行累积操作。
    maps:fold(
        fun % 定义一个匿名函数 (lambda表达式)
            %% 这个匿名函数有三个参数：Key, Value, AccumulatorIn (上一次迭代的结果)
            %% 模式匹配用于处理不同类型的Key
            (K, V0, AccIn) when K =:= peerhost; K =:= peername; K =:= sockname -> % 如果Key是网络相关地址
                AccIn#{K => ntoa(V0)}; % 使用ntoa转换后更新到累积器Map中
            ('User-Property', V0, AccIn) when is_list(V0) -> % 如果Key是'User-Property'且其值是一个列表 (MQTT 5.0 用户属性)
                AccIn#{ % 更新累积器Map
                    'User-Property' => maps:from_list(V0), % 将用户属性列表转换为Map
                    'User-Property-Pairs' => [ % 同时创建一个键值对列表的格式，方便某些场景使用
                        #{
                            key => Key,
                            value => Value
                        }
                        || {Key, Value} <- V0 % 列表推导，遍历V0中的每个键值对
                    ]
                };
            (_K, V, AccIn) when is_tuple(V) -> % 如果值是元组类型，通常是一些内部表示，这里选择忽略
                AccIn; % 不做任何修改，直接返回累积器
            (K, V, AccIn) -> % 处理其他所有类型的键值对
                AccIn#{K => V} % 直接将键值对添加到累积器Map中
        end,
        #{'User-Property' => #{}}, % 初始累积器Map，预先设置一个空的'User-Property'，确保该键总是存在
        Headers % 要遍历的原始Headers Map
    ).